package com.danan.realtime.app.dim;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.danan.realtime.app.BaseApp;
import com.danan.realtime.common.Constant;
import com.danan.realtime.pojo.TableProcess;
import com.danan.realtime.util.JDBCUtil;
import com.danan.realtime.util.SinkUtil;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @author NanHuang
 * @Date 2023/1/24
 */
public class DimApp extends BaseApp {
    public static void main(String[] args) {
        new DimApp().init(3001,3,"DimApp", Constant.TOPIC_ODS_DB);
    }
    @Override
    public void invoke(StreamExecutionEnvironment env, DataStreamSource<String> ds) {
        // 1 对数据进行ETL，过滤掉脏数据
        SingleOutputStreamOperator<JSONObject> etledDs = etl(ds);
        // 2 使用Flink CDC读取MySQL中配置表的信息
        SingleOutputStreamOperator<TableProcess> dimConfigDs = readDimConfig(env);
        // 3 根据配置信息在Phoenix中建表/删表
        SingleOutputStreamOperator<TableProcess> dimConfigDs1 = createTableInPhoenix(dimConfigDs);
        // 4 匹配数据和配置信息
        SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> dataAndConfigDs = matchDataAndConfig(etledDs, dimConfigDs1);
        // 5 将数据写入对应的维度表中
        dataAndConfigDs.addSink(SinkUtil.getPhoenixSink());

    }

    private SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> matchDataAndConfig(SingleOutputStreamOperator<JSONObject> data, SingleOutputStreamOperator<TableProcess> config) {
        // 1 将配置流设置为广播流
        MapStateDescriptor<String, TableProcess> desc = new MapStateDescriptor<>("dimConfig", String.class, TableProcess.class);
        BroadcastStream<TableProcess> dimConfig = config.broadcast(desc);
        // 2 数据流连接广播流
        return data.connect(dimConfig)
                .process(new BroadcastProcessFunction<JSONObject, TableProcess, Tuple2<JSONObject,TableProcess>>() {
                    @Override
                    public void processElement(JSONObject value, ReadOnlyContext ctx, Collector<Tuple2<JSONObject, TableProcess>> out) throws Exception {
                        // 1 获取广播状态
                        ReadOnlyBroadcastState<String, TableProcess> bs = ctx.getBroadcastState(desc);
                        // 2 根据key获取配置数据
                        String key = value.getString("table") + ":ALL";
                        TableProcess tp = bs.get(key);
                        if (tp != null) {
                            // 3 删除数据中不需要的字段
                            List<String> usefulColumns = Arrays.asList(tp.getSinkColumns().split(","));
                            JSONObject columns = value.getJSONObject("data");
                            columns.keySet().removeIf(k -> !usefulColumns.contains(k));
                            // 4 返回结果
                            out.collect(new Tuple2<>(columns,tp));
                        }
                    }

                    @Override
                    public void processBroadcastElement(TableProcess value, Context ctx, Collector<Tuple2<JSONObject, TableProcess>> out) throws Exception {
                        // 1 获取广播流
                        BroadcastState<String, TableProcess> bs = ctx.getBroadcastState(desc);
                        // 2 更新广播流数据
                        String key = value.getSourceTable() + ":" + value.getSourceType();
                        bs.put(key,value);
                    }
                });
    }

    private SingleOutputStreamOperator<TableProcess> createTableInPhoenix(SingleOutputStreamOperator<TableProcess> ds) {
        return ds.filter(t -> "dim".equals(t.getSinkType()))
                .process(new ProcessFunction<TableProcess, TableProcess>() {
                    private Connection conn;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        conn = JDBCUtil.getPhoenixConnection();
                    }

                    @Override
                    public void processElement(TableProcess in, ProcessFunction<TableProcess, TableProcess>.Context ctx, Collector<TableProcess> out) throws Exception {
                        switch (in.getOp()){
                            case "d":
                                dropTable(in,conn);
                                break;
                            case "u":
                                dropTable(in,conn);
                                createTable(in,conn);
                                break;
                            default:
                                createTable(in,conn);
                                break;
                        }
                        out.collect(in);
                    }


                    @Override
                    public void close() throws Exception {
                        conn.close();
                    }

                    private void dropTable(TableProcess in, Connection conn) throws SQLException {
                        System.out.println("Phoenix删表：" + in.getSinkTable());
                        // 1 拼接删表SQL
                        String sql = "drop table " + in.getSinkTable();
                        // 2 编译
                        PreparedStatement ps = conn.prepareStatement(sql);
                        // 3 执行
                        ps.execute();
                        // 4 关闭预编译
                        ps.close();
                    }


                    private void createTable(TableProcess in, Connection conn) throws SQLException {
                        System.out.println("Phoenix建表：" + in.getSinkTable());
                        // 1 拼接建表SQL
                        // 建表SQL模板
                        String template = "create table if not exists %s (" +
                                "   %s," +
                                "   constraint my_pk primary key (%s)" +
                                ") %s";
                        // 获取建表字段(字段名 + 字段类型)
                        String columns = in.getSinkColumns().replaceAll("[^,]+", "$0 varchar");
                        // 获取主键
                        String pk = in.getSinkPk() == null ? "id" : in.getSinkPk();
                        // 获取盐表(类似预分区)
                        String saltTable = in.getSinkExtend() == null ? "" : in.getSinkExtend();
                        // 填充建表SQL模板
                        String sql = String.format(template, in.getSinkTable(), columns, pk, saltTable);

                        // 2 预编译
                        PreparedStatement ps = conn.prepareStatement(sql);
                        // 3 执行
                        ps.execute();
                        // 4 关闭预编译
                        ps.close();
                    }
                });
    }

    private SingleOutputStreamOperator<TableProcess> readDimConfig(StreamExecutionEnvironment env) {
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("hadoop102")
                .port(3306)
                .databaseList("gmall_config")
                .tableList("gmall_config.table_process")
                .username("root")
                .password("123456")
                .deserializer(new JsonDebeziumDeserializationSchema())
                .build();
        return env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(),"mysql_cdc")
                .map(JSON::parseObject)
                .map(new RichMapFunction<JSONObject, TableProcess>() {
                    @Override
                    public TableProcess map(JSONObject in) throws Exception {
                        String op = in.getString("op");
                        TableProcess tp = null;
                        if ("d".equals(op)) {
                            tp = JSON.parseObject(in.getString("before"), TableProcess.class);
                        } else {
                            tp = JSON.parseObject(in.getString("after"), TableProcess.class);
                        }
                        tp.setOp(op);
                        return tp;
                    }
                });
    }

    private SingleOutputStreamOperator<JSONObject> etl(DataStreamSource<String> ds) {
        return ds.map(s -> s.replace("bootstrap-",""))
                .filter(new FilterFunction<String>() {
                    @Override
                    public boolean filter(String s) throws Exception {
                        try {
                            JSONObject in = JSON.parseObject(s);
                            return in.getString("database").equals("gmall2023")
                                    && (in.getString("type").equals("insert") || in.getString("type").equals("update"))
                                    && in.getJSONObject("data") != null
                                    && in.getString("table") != null;
                        } catch (Exception e) {
                            System.out.println("输入字符串格式有误，不能解析！（" + s + "）");
                            return false;
                        }
                    }
                })
                .map(JSON::parseObject);
    }
}
